package org.dkd.mq;

import javafx.scene.control.Label;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.dkd.controller.TestTableViewController;
import org.dkd.helper.SpringBootRegister;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.stream.Collectors;

/**
 * @className: DataCleanOrderKafkaConsumer
 * @Auther: charlie
 * @Date: 2023/2/23 18:53
 * @Description:
 */
@Component
@Slf4j
public class DataCleanOrderConsumer {


    /**
     * @auther: charlie
     * @date: 2023/2/15 15:58
     * @desc: 订单清洗 topic
     */
    @KafkaListener(
            topics = "test.abc.topic",
            groupId = "test.abc.topic.group",
            clientIdPrefix = "test.abc.topic-c-1"
            , concurrency = "1", containerFactory = "defaultContainerFactory",
            properties = {
                    ConsumerConfig.MAX_POLL_RECORDS_CONFIG + "=30",
                    ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG + "=600000"
            }
    )
    public void washOrderDay(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment acknowledgment) throws Exception {
        try {
            int orderNum = consumerRecords.size();
            log.info("清洗订单天的数据,batch->{}", orderNum);
            List<String> records = consumerRecords.stream().map(ConsumerRecord::value).collect(Collectors.toList());
            for (String record : records) {
                String[] arr = record.split(":");
                String timestamp = arr[1];
                long delay = System.currentTimeMillis() / 1000 - Long.parseLong(timestamp) / 1000;
                System.out.println(new Label(record + "    delay:" + delay + "s"));
            }
            acknowledgment.acknowledge();
        } catch (Exception e) {
            e.printStackTrace();
            throw e;
        }
    }
}
